#include <amqp.h>
#include "head.h"
#include "rabbitmq_api.h"

extern struct Config config;

int create_queue(key_t key) {
    int qid;
    if ((qid = msgget(key, IPC_CREAT | 0666)) < 0) {
        return -1;
    }
    
    return qid;
}

int send_to_queue(int qid, int type) {
    struct msgbuf msg;
    msg.mtype = type;
    if (msgsnd(qid, (void *)&msg, sizeof(msg.mtype), IPC_NOWAIT) < 0) {
        return -1;
    }
    return 0;
}

int recv_from_queue(int qid) {
    struct msgbuf msg;
    bzero(&msg, sizeof(msg));
    if (msgrcv(qid, (void *)&msg, sizeof(msg.mtext), 0, 0) < 0) {
        return -1;
    }
    return msg.mtype;
}

void *thread_send_msg(void *arg) {
    pthread_detach(pthread_self());
    struct thread_sndMsg sndMsg = *(struct thread_sndMsg *)arg; 
    // DBG("<"L_RED"Dubug"NONE"> : send msg thread created for system detect...");
    // DBG("<"L_RED"Dubug"NONE"> : type=%d; time=%d", sndMsg.type, sndMsg.time);
    while (1) {
        send_to_queue(sndMsg.qid, sndMsg.type);
        sleep(sndMsg.time);
    }
}

void *thread_recv_msg(void *arg) {
    // DBG("<"L_BLUE"INFO"NONE"> : In recv msg thread <%ld>", pthread_self());
    pthread_detach(pthread_self());
    struct threadpool_t *pool = (struct threadpool_t *)arg;
    FILE *fp = NULL;
    char buff[1024] = {0};
    // amqp_connection_state_t conn = connectServer(pool->channel);
    // amqp_channel_open(pool->conn, pool->channel);
    // amqp_rpc_reply_t replyRes = amqp_get_rpc_reply(pool->conn);
    // if (replyRes.reply_type != AMQP_RESPONSE_NORMAL) {
    //     DBG("<"RED"ERROR"NONE"> open channel failed, <%ld>thread exit!", pthread_self());
    //     return NULL;
    // }
    amqp_connection_state_t conn = connectServer("47.111.117.240", 5672);
    amqp_channel_open(conn, 1);

    while (1) {
        pthread_mutex_lock(&(pool->lock));
        int type = recv_from_queue(pool->qid);
        if (type < 0) {
            fprintf(stderr, "recv msg from queue failed\n");
            pthread_mutex_unlock(&(pool->lock));
            sleep(2);
            continue;
        } 
        // DBG("<"L_BLUE"INFO"NONE"> : <%ld> type=%d", pthread_self(), type);
        pthread_mutex_unlock(&(pool->lock));
        switch(type) {
        //     case 1:
        //         fp = popen("bash ./script/CpuLog.s", "r");
        //         break;
            case 2:
                fp = popen("bash ../script/Memlog.sh", "r");
                break;
        //     case 3:
        //         fp = popen("bash ./script/DiskLog.s", "r");
        //         break;
        //     case 4:
        //         fp = popen("bash ./script/ProcLog.s", "r");
        //         break;
        //     case 5:
        //         fp = popen("bash ./script/SysLog.s", "r");
        //         break;
            case 6:
                fp = popen("bash ../script/UserLog.sh", "r");
                break;
            default:
                continue;
                break;
        }
        if (fp == NULL) {
            perror("popen");
            continue;
        }
        size_t rsize = fread(buff, 1, sizeof(buff), fp);
        if (rsize < 0) {
            perror("fread");
        }
        // publishMessage(pool->conn, pool->channel, type, config.hostName, config.ipAddress, buff);
        publishMessage(conn, 1, type, config.hostName, config.ipAddress, buff);
        pclose(fp);
        memset(buff, 0, sizeof(buff));
        sleep(2);
    }
}

void send_loop(int qid, struct Config *config) {
    pthread_t tid[SEND_THREAD];
    struct thread_sndMsg sndJob[SEND_THREAD]; // = (calloc(SEND_THREAD, sizeof(struct thread_sndMsg);
    for (int i = 0; i < SEND_THREAD; i++) {
        sndJob[i].qid = qid;
        switch (i)
        {
        case 0:
            sndJob[i].time = config->interTime.cpu.time;
            sndJob[i].type = config->interTime.cpu.type;
            break;
        case 1:
            sndJob[i].time = config->interTime.disk.time;
            sndJob[i].type = config->interTime.disk.type;
            break;
        case 2:
            sndJob[i].time = config->interTime.mem.time;
            sndJob[i].type = config->interTime.mem.type;
            break;
        case 3:
            sndJob[i].time = config->interTime.proc.time;
            sndJob[i].type = config->interTime.proc.type;
            break;
        case 4:
            sndJob[i].time = config->interTime.sys.time;
            sndJob[i].type = config->interTime.sys.type;
            break;
        case 5:
            sndJob[i].time = config->interTime.user.time;
            sndJob[i].type = config->interTime.user.type;
            break;
        default:
            break;
        }
        // DBG("child thread will be created; type:%d, time:%d\n", sndJob[i].type, sndJob[i].time);
        pthread_create(&tid[i], NULL, thread_send_msg, (void *)&sndJob[i]);
    }
    sleep(10);
    // DBG("<"YELLOW"Debug"NONE"> : end");
}